/* 
 * File:   thread_group.hpp
 * Author: jdkunk
 *
 * Created on April 10, 2011, 12:35 PM
 */

#ifndef THREAD_GROUP_HPP
#define	THREAD_GROUP_HPP

#include <vector>
#include <list>
#include <boost/thread.hpp>

#ifdef BOOST_MSVC
#pragma warning(push)
#pragma warning(disable:4251)
#endif

namespace jdkunk
{
    using namespace boost;
    
    class thread_group
    {
    private:
        
        thread_group(thread_group const&);
        thread_group& operator=(thread_group const&); 
        
        template< typename F >
        void reader_iterator_method( F f ) {
            boost::shared_lock<shared_mutex> guard(m);
            unsigned last_written = time_stamp;
            std::list<thread*> threads = this->threads;
            guard.unlock();
            
            for(std::list<thread*>::iterator it=threads.begin(),end=threads.end();
                it!=end;
                ++it)
            {
                boost::shared_lock<shared_mutex> inner_guard(m);
                if( time_stamp > last_written ) {
                  last_written = time_stamp;
                  
                  std::vector<thread*> add;
                  std::vector<std::list<thread*>::iterator> remove;
                
                  std::list<thread*>::iterator jt, je;
                  
                  // For each t in T,
                  //   if t does not exists in T', 
                  //       remove it;
                  // For each t in T',
                  //   if t does not exists in T,
                  //       add it;
                  // where T equals local threads
                  // and T' equals class threads.
                  
                  for(jt=threads.begin(),je=threads.end();
                      jt!=je; ++jt)
                  {
                    if( this->threads.end() == 
                    std::find( this->threads.begin(), this->threads.end(),*jt ))
                      remove.push_back(jt);
                  }
                  
                  for(jt=this->threads.begin(),je=this->threads.end();
                      jt!=je; ++jt)
                  {
                    if( threads.end() == 
                    std::find( threads.begin(), threads.end(),*jt ) ) {
                      add.push_back(*jt);
                    }
                  }
                  
                  for( std::vector<std::list<thread*>::iterator>::iterator 
                    ktr = remove.begin(), kend = remove.end(); 
                    ktr != kend; ++ktr ) 
                  {
                    threads.erase(*ktr);
                  }
                  
                  for( std::vector<thread*>::iterator st = add.begin(), se = add.end(); 
                    st != se; ++st ) 
                      threads.push_back(*st);
                }
                inner_guard.unlock();
                
                if( *it ) {
                    if( f(*it) )
                      return;
                }
            }
        }
        
        static bool join_all_callback(boost::thread * t) {
          t->join();
          return false;
        }
        static bool interrupt_all_callback(boost::thread * t) {
          t->interrupt();
          return false;
        }
        
        struct is_my_thread {
          thread * t; 
          is_my_thread( thread * t ) : t(t) {}
          bool operator()(thread * r) { return t==r; }
        };
        
        template< typename P >
        struct find_thread_if_callback {
            P p; thread * & t;
            find_thread_if_callback(P p, thread * & t) : p(p), t(t) {}
            bool operator()(thread * r) { 
              bool result = (t==r);
              if(result)
                t = r;
              return result;
            }
        };
        
        template< typename P >
        struct return_false {
            P p; 
            return_false(P p) : p(p) {}
            bool operator()(thread * r) { 
              p(r);
              return false;
            }
        };
    public:
        thread_group() : time_stamp(0) {}
        ~thread_group()
        {
            for(std::list<thread*>::iterator it=threads.begin(),end=threads.end();
                it!=end;
                ++it)
            {
                delete *it;
            }
        }

        template<typename F>
        thread* create_thread(F threadfunc)
        {
            boost::lock_guard<shared_mutex> guard(m);
            thread * t = new thread(threadfunc);
            threads.push_back(t);
            return t;
        }
        
        void add_thread(thread* thrd)
        {
            if(thrd)
            {
                boost::shared_lock<shared_mutex> shared(m);
                if( threads.end() == 
                std::find( threads.begin(), threads.end(), thrd ) ) {
                    boost::upgrade_lock<shared_mutex> guard(m);
                    threads.push_back(thrd);
                }
            }
        }
        
        void join_all()
        {
            reader_iterator_method(join_all_callback);
        }
        
        void interrupt_all()
        {
            reader_iterator_method(interrupt_all_callback);
        }
        
        template<typename P>
        thread * find_thread_if(P p) {
            thread * t = NULL;
            find_thread_if_callback<P> ft(p,t);
            reader_iterator_method( ft );
        }
        
        template<typename P>
        void for_each_thread(P p) {
            reader_iterator_method( return_false<P>(p) );
        }
        
        template<typename P>
        void remove_thread_if(P p) {
          bool changed = false;
          boost::lock_guard<shared_mutex> guard(m);
          
          for(std::list<thread*>::iterator it=threads.begin(),end=threads.end();
              it!=end;
              ++it)
          {
            if( p(*it) ) {
              changed = true;
              threads.erase(it);
            }
          }
          if( changed )
            ++time_stamp;
        }
            
        void remove_thread(thread* thrd)
        {
            remove_thread_if( is_my_thread(thrd) );
        }
        
        template<typename P>
        void delete_thread_if(P p) {
          bool changed = false;
          boost::lock_guard<shared_mutex> guard(m);
          
          for(std::list<thread*>::iterator it=threads.begin(),end=threads.end();
              it!=end;
              ++it)
          {
            if(p(*it)) {
              changed = true;
              boost::thread * t = *it;
              threads.erase(it);
              delete t;
            }
          }
          if( changed )
            ++time_stamp;
        }
        
        void delete_thread(thread * t) {
          delete_thread_if( is_my_thread(t) );
        }
        
        size_t size() const
        {
            boost::shared_lock<shared_mutex> guard(m);
            return threads.size();
        }
        
    private:
        volatile unsigned time_stamp;
        std::list<thread*> threads;
        mutable shared_mutex m;
    };
}

#ifdef BOOST_MSVC
#pragma warning(pop)
#endif

#endif	/* THREAD_GROUP_HPP */

